package main

import (
	"github.com/Shopify/sarama"
	"github.com/astaxie/beego/logs"
)

var client sarama.SyncProducer

func initKafka() (err error) {
	co := sarama.NewConfig()
	co.Producer.RequiredAcks = sarama.WaitForAll
	co.Producer.Partitioner = sarama.NewRandomPartitioner
	co.Producer.Return.Successes = true

	target := []string{"localhost:7777", "localhost:8888"}
	client, err = sarama.NewSyncProducer(target, co)
	if err != nil {
		logs.Error(err)
	}
	defer client.Close()
	return
}

func sendToKafka(data *TextMsg) (err error) {
	err = initKafka()
	msg := &sarama.ProducerMessage{}
	msg.Topic = data.Topic
	msg.Value = sarama.StringEncoder(data.Msg)
	_, _, err = client.SendMessage(msg)
	if err != nil {
		logs.Error("send message failed,err:%v data %v topic:%v", err, data.Msg, &data.Topic)
	}
	logs.Debug("send message succ: data %v topic:%v", data.Msg, &data.Topic)
	return
}
